package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.RedisCacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@Service
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;

    @Autowired
    private RedisCacheService redisCacheService;

    @Autowired
    private ThreadPoolTaskExecutor mythreadPool;
    /**
     * 添加任务
     *
     * @param task 任务对象
     * @return 任务id
     */
    @Override
    public void addTask(Task task) {

        mythreadPool.execute(new Runnable() {
            @Override
            public void run() {
                //1.任务添加到数据库中
                Boolean flag = saveTaskToDB(task);

                if(flag){
                    saveTaskToRedis(task);
                }
            }
        });


    }



    /**
     * 保存任务到redis中
     * @param task
     */
    private void saveTaskToRedis(Task task) {
        //2.添加任务到redis中
        String keySuffix = task.getTaskType() + "_" + task.getPriority();
        //获取未来的五分钟时间
        Calendar calendar=Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);//获取未来时间5分钟
        long timeInMillis = calendar.getTimeInMillis();

        //2.1 如果任务的执行时间小于等于当前系统时间，则存入到list队列中
        if(task.getExecuteTime()<=System.currentTimeMillis()){
            redisCacheService.lLeftPush(ScheduleConstants.TOPIC+keySuffix, JSON.toJSONString(task));
        }else
            if(task.getExecuteTime()>System.currentTimeMillis() && task.getExecuteTime()<=timeInMillis){
                //2.2 如果任务的执行时间大于当前系统时间，小于预设时间(5分钟)，则存入到zset队列中
                redisCacheService.zAdd(ScheduleConstants.FUTURE+keySuffix, JSON.toJSONString(task), task.getExecuteTime());
        }
    }

    /**
     * 保存任务到数据库表中
     * @param task
     * @return
     */
    private Boolean saveTaskToDB(Task task) {
        //1.1 保存数据到taskinfo表中
        Taskinfo taskinfo=new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskinfo);

        //为Task类中的id赋值
        task.setTaskId(taskinfo.getTaskId());


        //1.2 保存的数据到taskinfologs表中
        TaskinfoLogs taskinfologs=new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfologs);
        taskinfologs.setVersion(1);
        taskinfologs.setStatus(0);
        int insert = taskinfoLogsMapper.insert(taskinfologs);
        return insert==1;
    }


    /**
     * 拉取任务，消费任务，直接从redis中操作
     *
     * @param taskType 类型
     * @param priority 优先级
     * @return
     */
    @Override
    public Task pullTask(Integer taskType, Integer priority) {

        Future<Task> future = mythreadPool.submit(new Callable<Task>() {
            @Override
            public Task call() throws Exception {
                System.out.println("开始进行拉取任务了....");
                //定义key
                String keySuffix = taskType+ "_" + priority;
                //1.从redis中的list队列中取出数据，并从redis中删除
                String taskStr = redisCacheService.lRightPop(ScheduleConstants.TOPIC + keySuffix);
                Task task = JSON.parseObject(taskStr, Task.class);
                if(task!=null){
                    //2.删除任务表
                    taskinfoMapper.deleteById(task.getTaskId());


                    //3.修改任务日志表
                    TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(task.getTaskId());
                    if(taskinfoLogs!=null){
                        //按道理来说要修改vesion,version=version+1,但是我们已经集成mp集成乐观锁
                        //修改状态
                        taskinfoLogs.setStatus(ScheduleConstants.EXECUTED);
                        taskinfoLogsMapper.updateById(taskinfoLogs);
                    }
                }
                System.out.println("拉取任务结束了....");
                //返回数据
                return task;
            }
        });
        try {
            return future.get();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 定时任务执行刷新同步数据从zset中到list中
     * 没有别人调用该方法，所以说没有参数，没有返回值
     * 暂定每分钟执行一次
     */
    @Override
   // @Scheduled(cron = "0 0/1 * * * ? ")
    @Scheduled(cron = "*/1 * * * * ?")//暂定每秒钟执行一次，方便测试
    public void refresh() {

        mythreadPool.execute(new Runnable() {
            @Override
            public void run() {
                // String token = redisCacheService.tryLock("FUTURE_TASK_SYNC", 30 * 1000);
                //表示已经拿到锁，可以执行业务
                // if(StringUtils.isNotBlank(token)){
                System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
                System.out.println(new DateTime());

                try {
                    //加入休眠时间
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //1.获取未来数据的所有key
                Set<String> futureKeys = redisCacheService.scan(ScheduleConstants.FUTURE + "*");
                if(futureKeys!=null && futureKeys.size()>0){
                    for (String futureKey : futureKeys) {
                        //根据key获取对应的value,当前数据都是到期了
                        Set<String> futureValues = redisCacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());

                        //3.同步数据到list中
                        String keysuffix=futureKey.split(ScheduleConstants.FUTURE)[1];
                        String topicKey=ScheduleConstants.TOPIC+keysuffix;
                        redisCacheService.refreshWithPipeline(futureKey,topicKey,futureValues);
                        System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
                    }
                }
                //}
            }
        });



    }

    /**
     * 定时同步数据，从mysql中到redis中
     * 没有方法和返回值，也是不会被别的接口调用
     * 暂定每5分钟执行一次
     */
    @Override
    @Scheduled(cron = "0 0/5 * * * ? ")
    public void syncTaskToRedis() {
        //1.清除redis中缓存
        Set<String> listkeys = redisCacheService.scan(ScheduleConstants.TOPIC + "*");
        Set<String> zsetkeys = redisCacheService.scan(ScheduleConstants.FUTURE + "*");
        redisCacheService.delete(listkeys);
        redisCacheService.delete(zsetkeys);
        //2.从mysql表中查询执行时间小于等于未来五分钟的数据
        Calendar calendar=Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);//获取未来时间5分钟
        List<Taskinfo> taskinfoList = taskinfoMapper.selectList(
                Wrappers.<Taskinfo>lambdaQuery().le(Taskinfo::getExecuteTime, calendar.getTimeInMillis())
        );
        if(taskinfoList!=null && taskinfoList.size()>0){
            for (Taskinfo taskinfo : taskinfoList) {
                //3.把数据保存到redis中
                Task task=new Task();
               BeanUtils.copyProperties(taskinfo,task);
               task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                this.saveTaskToRedis(task);
            }
        }
    }
}
